package com.mq.core.container;

import com.alibaba.fastjson.JSON;
import com.mq.core.Message;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.data.redis.core.ZSetOperations;

import java.util.Set;
import java.util.concurrent.*;

public class SimpleMessageListenerContainer extends AbstractMessageListenerContainer {

    private static final Log logger = LogFactory.getLog(SimpleMessageListenerContainer.class);

    private String delayedSuffix = "-delayed";
    private Executor executor;
    private boolean isRunning;

    @Override
    public void doStart() {
        if(isRunning){
            return;
        }
        isRunning = true;
        for(String queueName : getQueueName()){
            executor = Executors.newFixedThreadPool(2);
            executor.execute(new AsyncConsumer(queueName));
            executor.execute(new AsyncConsumerDelayed(queueName));
        }
    }

    /**
     * 消费队列
     */
    private class AsyncConsumer implements Runnable {

        private String queueName;
        private String delayedQueueName;

        public AsyncConsumer(String queueName){
            this.queueName = queueName;
            this.delayedQueueName = queueName+delayedSuffix;
        }

        @Override
        public void run() {
            int maxConsumer = getMaxConsumer();
            int retryTimes = getRetryTimes();
            ExecutorService consumer = new ThreadPoolExecutor(1, getMaxConsumer(),
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>());
            while(true){
                try {
                    if(isQuit()){
                        Thread.sleep(3000);
                        continue;
                    }
                    // 1.取队列信息
                    final String msg = getRedisTemplate().opsForList().rightPop(queueName);
                    if (msg != null) {
                        // 2.从json对象中解码出通知信息
                        final Message message = JSON.parseObject(msg, Message.class);

                        consumer.execute(()->{
                            try{
                                invoke(message);
                            }catch (Exception e){
                                logger.error(String.format("队列=%s消费失败, msg=%s", queueName, msg), e);
                                //如果小于最大重试次数 进入死信队列
                                if(message.getRetryTimes() < getRetryTimes()){
                                    double score = System.currentTimeMillis() + (3 * 60 * 1000);
                                    message.setRetryTimes(message.getRetryTimes()+1);
                                    getRedisTemplate()
                                            .opsForZSet().add(delayedQueueName, JSON.toJSONString(message), score);
                                }
                            }
                        });
                    } else {
                        Thread.sleep(1000);
                    }
                } catch (Exception e) {
                    logger.error(e);
                    logger.info(String.format("consumer error queues = %s, error = %s", queueName, e.getMessage()));
                }
            }
        }
    }

    /**
     * 死信队列
     */
    private class AsyncConsumerDelayed implements Runnable {

        private String queueName;
        private String delayedQueueName;

        public AsyncConsumerDelayed(String queueName){
            this.queueName = queueName;
            this.delayedQueueName = queueName+delayedSuffix;
        }

        @Override
        public void run() {
            while (true) {
                try {
                    if(isQuit()){
                        Thread.sleep(3000);
                        continue;
                    }
                    // 1.取出准备就绪的消息内容
                    ZSetOperations<String, String> vo = getRedisTemplate().opsForZSet();
                    Set<ZSetOperations.TypedTuple<String>> items = vo.rangeWithScores(delayedQueueName, 0, 0);
                    ZSetOperations.TypedTuple<String> item = items.size() > 0 ? items.iterator().next() : null;
                    if(item == null || item.getScore() > System.currentTimeMillis()){
                        try {
                            Thread.sleep(100);
                            continue;
                        }catch (Exception e){
                            logger.error(e);
                        }
                    }
                    // 2.解析消息内容
                    String msg = item.getValue();
                    // 3.转移到就绪队列
                    Long rSize = vo.remove(delayedQueueName, msg);
                    if(rSize != null && rSize.intValue() > 0){
                        getRedisTemplate().opsForList().leftPush(queueName, msg);
                    }
                }catch (Exception e){
                    logger.error(e.getMessage(), e);
                }
            }
        }
    }
}
